package com.mall.promo.mq;

import com.alibaba.fastjson.JSON;
import com.mall.order.dto.CreateSeckillOrderRequest;
import com.mall.promo.cache.CacheManager;
import com.mall.promo.dal.persistence.PromoItemMapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.redisson.api.RLock;
import org.redisson.api.RMap;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.nio.charset.Charset;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.locks.Lock;

/**
 * 1.com.mall.promo.service.PromoServiceImpl中调用sendPromoTransactionMessage，发送事务消息
 * 2.sendPromoTransactionMessage调用transactionMQProducer中的sendMessageInTransaction，传入message,arg参数，执行事务消息的发送---->事务消息发送成功，RocketMQ会响应，消息发送成功
 * 3.监听器调用：executeLocalTransaction，执行本地事务:扣减库存---->将执行结果，存储在redis中，被checkLocalTransaction，检查
 * 4.监听器调用：checkLocalTransaction，执行本地事务监听,检查
 */
@Component
@Slf4j
public class CreatePromoOrderProducer_X {

    private TransactionMQProducer transactionMQProducer;  //该对象不是我们自己定义的，管理的；没有被@Service ，不会被spring容器管理；所以，我们应该：
    //1. 定义init()方法；2.@PostConstruct注解---->使得spring容器可以调用init方法对TransactionMQProducer进行初始化

    @Autowired
    PromoItemMapper itemMapper;   //执行扣减库存操作的一个工具

    @Autowired
    CacheManager cacheManager;  //存储本地事务执行的结果，便于：后续“检查事务”可以获取到本地事务执行的结果

    @Autowired
    RedissonClient redissonClient;    //从redis中获取一把分布式锁


    @PostConstruct
    public void init() {
        log.info("CreatePromoOrderProducer init start ...");
        //1.创建事务的produer对象
        transactionMQProducer = new TransactionMQProducer("create_promo_order_producer");
        //2.设置发送地址
        transactionMQProducer.setNamesrvAddr("127.0.0.1:9876");
        //3.设置监听器：（1）执行本地方法；（2）检查本地事务执行结果
        transactionMQProducer.setTransactionListener(new TransactionListener() {
            /**
             * 执行本地事务----执行秒杀服务的本地事务：扣减秒杀商品的库存
             * @param msg
             * @param arg
             * @return
             */
            @Override
            public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
                //执行秒杀服务的本地事务：扣减秒杀商品的库存

                //接收arg这个由producer发送的map参数
                Map<String, Long> paraMap = (Map<String, Long>) arg;
                Long productId = paraMap.get("productId");
                Long psId = paraMap.get("psId");
                Long userId = paraMap.get("userId");

                /**
                 * 扣减库存的时候，先验证：用户是否已经购买
                 */
//                Map<Long, Boolean> mapY = redissonClient.getMap(productId.toString());   //一个关于productId的hash表
//                mapY.put(userId, 1);   //value为1：用户有一次机会购买

                //这种方式，不可取
//                synchronized (this){
//                Integer integer = itemMapper.decreaseStock(productId, psId);//该方法需要传入：productId和psId--->从transactionMQProducer.sendMessageInTransaction的参数中提取
//                }

                //对秒杀活动中，不同场次的不同商品，拥有唯一id，根据唯一id，进行上锁
                String lockKey = "promo_order_lock" + productId + "--" + psId;
                Lock stockLock = redissonClient.getLock(lockKey);
                stockLock.lock();  //上锁--->如果加锁失败，当前线程会阻塞等待
                Integer integer = null;//该方法需要传入：productId和psId--->从transactionMQProducer.sendMessageInTransaction的参数中提取
                try {   //仅仅同一个商品，对应的userId的value值：为1的时候可以购买
//                    Boolean result = mapY.get(userId);
//                    if (result == false) {
                    integer = itemMapper.decreaseStock(productId, psId);
//                        return LocalTransactionState.ROLLBACK_MESSAGE;
//                    }
////                    else {
////                        return LocalTransactionState.ROLLBACK_MESSAGE;
////                    }
                } finally {
                    stockLock.unlock();  //开锁---》保证一定可以开锁，使用finally
                }
                /**
                 * 思考：本地事务执行完毕，执行的结果；怎么可以让“checkLocalTransaction”检查本地事务查询到？
                 * 可以存储在mysql中，也可以存储在redis中；这里存储在redis中
                 */
                String transactionKey = "promo_local_transaction" + msg.getTransactionId();   //根据本地事务的唯一id，搭配本地事务的执行结果，存储在redis中

                String promoStockKey = "promo_order_stock_not_enough" + "-" + productId + "-" + psId;
                if (integer < 1) {
                    cacheManager.setCache(transactionKey, "fail", 1);   //将本地事务执行失败的结果，存储在redis中

                    /***
                     * 秒杀商品售罄---打标记---商品售罄
                     */

                    cacheManager.setCache(promoStockKey,"none",1);  //过期时间：1天

                    return LocalTransactionState.ROLLBACK_MESSAGE;  //没有库存可以扣减
                }
//                mapY.put(userId, true);

//                /**
//                 *对秒杀成功用于进行标记
//                 */
//                Map<String, Long> map = redissonClient.getMap(userId.toString());
//                map.put("productId", productId);

                cacheManager.setCache(transactionKey, "success", 1);   //将本地事务执行成功的结果，存储在redis中
                return LocalTransactionState.COMMIT_MESSAGE;  //扣减成功
            }

            /**
             * 检查本地事务执行结果
             * @param msg
             * @return
             */
            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt msg) {
                //1.根据本地事务唯一的id，从redis中，获取到本地事务执行的结果并做出反应
                String transactionKey = "promo_local_transaction" + msg.getTransactionId();
                String result = cacheManager.checkCache(transactionKey);

                //2.如果result为null或者空字符串，则意味着：本地事务可能暂时没有执行完；先返回UNKNOW
                if (result == null || result.trim().equals("")) {
                    return LocalTransactionState.UNKNOW;
                }
                if ("success".equals(result)) {   //本地事务执行成功
                    return LocalTransactionState.COMMIT_MESSAGE;
                }
                return LocalTransactionState.ROLLBACK_MESSAGE;   //本地事务执行失败
            }
        });
        //4.启动
        try {
            transactionMQProducer.start();
            log.info("CreatePromoOrderProducer init end ...");
            return;
        } catch (MQClientException e) {
            e.printStackTrace();
        }
        log.info("CreatePromoOrderProducer init failed ...");
    }


    /**
     * 提供给PromoServiceImpl进行调用，发送事务消息的方法
     *
     * @param request
     * @param productId
     * @param psId
     * @return 注：返回值：boolean 如果为true：消息发送成功并且事务执行成功   false：有任意一个执行失败
     */
    public boolean sendPromoTransactionMessage(CreateSeckillOrderRequest request, Long productId, Long psId, Long userId) {  //productId和psId
        //1.接收消息(将JavaBean对象转化为string，后续再转化为字节数据)
        String data = JSON.toJSONString(request);  //ali开源的JSON工具
        //2.准备待发送消息
        Message message = new Message();
        message.setTopic("create_promo_order");
        message.setBody(data.getBytes(Charset.forName("utf-8")));  //string转化为字节数据

        /**
         * 补充：对productId和psId的额外封装，封装到：sendMessageInTransaction的arg参数中
         */
        Map<String, Long> arg = new HashMap<>();  //使用map封装
        arg.put("productId", productId);
        arg.put("psId", psId);
        arg.put("userId", userId);

        //3.发送消息
        TransactionSendResult sendResult = null;//第二个参数arg将交给executeLocalTransaction
        try {
            sendResult = transactionMQProducer.sendMessageInTransaction(message, arg);  //arg发送给本地事务executeLocalTransaction执行

        } catch (MQClientException e) {
            e.printStackTrace();
        }

        System.out.println(sendResult.getLocalTransactionState());

        //4.根据返回结果，进行响应
        if (sendResult != null && LocalTransactionState.COMMIT_MESSAGE.equals(sendResult.getLocalTransactionState())) {
//
//            /**
//             *对秒杀成功用于进行标记
//             */
//            Map<String, Long> map = redissonClient.getMap(userId.toString());
//            map.put("productId", productId);

            return true;   //表示发送消息成功；并且本地事务执行成功
        }
        return false;
    }
}
